package com.bw.ad1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.bean.CitySort;
import com.bw.bean.OrderStatus;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hdfs.util.Diff;

import java.text.SimpleDateFormat;
import java.util.*;

public class Test11 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9", "test4" + System.currentTimeMillis()));

        //筛选订单数据
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String tableName = jsonObject.getString("table");
                if ("order_info".equals(tableName)) {
                    String status = jsonObject.getJSONObject("data").getString("status");
                    if ("60010".equals(status) || "60030".equals(status)) {
                        return true;
                    }
                    return false;
                }
                return false;
            }
        });
//        orderInfoDS.print("stream>>");

        // 变成JSON数据
        SingleOutputStreamOperator<JSONObject> mapStream = orderInfoDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        });
        // 添加水位线
        SingleOutputStreamOperator<JSONObject> watermarkStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                // 时间戳13位
                return element.getJSONObject("data").getLong("create_time") ;
            }
        }));


        // 根据发送者城市进行key
        SingleOutputStreamOperator<CitySort> processStream = watermarkStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("data").getString("sender_city_id");
            }
        }).window(TumblingEventTimeWindows.of(Time.minutes(60))).process(new ProcessWindowFunction<JSONObject, CitySort, String, TimeWindow>() {
            private MapState<String, String> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("map_state", String.class, String.class));
            }

            @Override
            public void process(String key, ProcessWindowFunction<JSONObject, CitySort, String, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<CitySort> collector) throws Exception {
                /*
                    city_id  id  status    update_time
                        1    1     60010        30          --- 30 下单
                        1    1     60030        50          --- 50 取件
                        1    2     60010        40          --- 40 下单
                        1    3     60010        20          ---20 下单
                        1    2     60030        80          ---30 取件

                        id,
                 */
//                System.out.println("key = " + key);
                int all_diff = 0;
                int count = 0;
                Iterator<JSONObject> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    JSONObject orderInfo = iterator.next();
                    // 获取订单ID
                    String ID = orderInfo.getJSONObject("data").getString("id");
                    String create_time = orderInfo.getJSONObject("data").getString("create_time");
                    // 取件时间
                    String update_time = orderInfo.getJSONObject("data").getString("update_time");

                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                    sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));


                    // 取出时间
                    String value = mapState.get(ID);
                    if (value != null) {
                        // 放下单的时间
                        mapState.put(ID, create_time);
                    } else {
                        // value 本身就是下单时间
                        long create_time_ts = sdf.parse(create_time).getTime();
                        long update_time_ts = sdf.parse(update_time).getTime();
                        long diff = update_time_ts - create_time_ts;
                        all_diff += diff;
                        count++;
                    }
                }
                // 平均时长
                double avg_ts = (double) all_diff / count;

                CitySort citySort = new CitySort();
                citySort.setCityId(key);
                citySort.setAvgTs(avg_ts);
                citySort.setEdt(context.window().getEnd());
                collector.collect(citySort);

            }
        });
//        processStream.print("processStream>>");
        /*
            city_id  avg_ts
                1     20
                2     30
                3     5
                4     80



                1   10
                2   30
                5   60
         */
        // 根据窗口在进行keyby
        SingleOutputStreamOperator<String> triggerTSStream = processStream.keyBy(new KeySelector<CitySort, String>() {
            @Override
            public String getKey(CitySort citySort) throws Exception {
                return citySort.getEdt() + "";
            }
        }).process(new KeyedProcessFunction<String, CitySort, String>() {
            private ListState<CitySort> CitySorts;
            private ValueState<Long> triggerTS;

            @Override
            public void open(Configuration parameters) throws Exception {
                triggerTS = getRuntimeContext().getState(new ValueStateDescriptor<Long>("triggerTS", Long.class));
                CitySorts = getRuntimeContext()
                        .getListState(new ListStateDescriptor<CitySort>("hotItems", CitySort.class));
            }

            @Override
            public void processElement(CitySort citySort, KeyedProcessFunction<String, CitySort, String>.Context context, Collector<String> collector) throws Exception {
                CitySorts.add(citySort);
                if (triggerTS.value() == null) {
                    context.timerService().registerEventTimeTimer(citySort.getEdt() + 1L);
                    triggerTS.update(citySort.getEdt());
                }
            }

            // 定时器触发方法
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, CitySort, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                // 拿出状态里面的所有数据
                Iterable<CitySort> cs = this.CitySorts.get();

                // 存储最终的结果
                ArrayList<CitySort> result = new ArrayList<>();


                for (CitySort citysort : cs) {
                    result.add(citysort);
                }
                // 清空状态
                this.CitySorts.clear();
                // 清空定时器
                triggerTS.clear();

                // 对result 排序取前3
                result.sort((o1, o2) -> o1.getAvgTs().intValue() - o2.getAvgTs().intValue());


                // 拼出最后的显示结果
                StringBuilder sb = new StringBuilder();
                sb.append("窗口结束时间: " + (timestamp - 1) + "\n");
                sb.append("---------------------------------\n");
                for (int i = 0; i < result.size(); i++) {
//                    sb.append("排名:"+(i+1)+result.get(i) + "\n");
                    sb.append("排名"+(i+1)+"城市"+result.get(i).getCityId()+"用时"+result.get(i).getAvgTs() + "\n");
                }
//                for (int i = 0; i < 3; i++) {
//                    sb.append(result.get(i) + "\n");
//                }
                sb.append("---------------------------------\n\n");
                out.collect(sb.toString());

            }
        });

        triggerTSStream.print("processStream>>");

        env.execute();
    }
}
